package com.project.KafkaRoute.executor;

import com.alibaba.fastjson.JSONObject;
import com.project.MessageRoute.executor.ProductExecutor;
import com.project.MessageRoute.model.MessageValueEntity;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Component;

import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ExecutionException;

@Primary
@Component
public class KafkaProductExecutor implements ProductExecutor {

    private volatile boolean showdown = false;

    private KafkaProducer<String, String> producer;

    @Value("${kafka.servers:}")
    private String servers;

    protected synchronized void init(){
        if (null == producer) {
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            props.put(ProducerConfig.ACKS_CONFIG, "1");

            producer = new KafkaProducer<>(props);
        }
    }

    @Override
    public void shutdown() {
        if (!showdown) {
            KafkaPushBinder.consumers.forEach(f->f.close());
            showdown = true;
        }
    }

    @Override
    public void sendGroup(String groupId, String topic, String tags, String... jsons) {
        if (StringUtils.isEmpty(tags)) {
            return;
        }

        init();

        MessageValueEntity messageValueEntity = new MessageValueEntity();
        messageValueEntity.setParameter(new String[jsons.length]);
        messageValueEntity.setTaskId(UUID.randomUUID().toString());
        for(int i=0;i<jsons.length;i++){
            messageValueEntity.getParameter()[i] = jsons[i];
        }
        String json = JSONObject.toJSONString(messageValueEntity);

        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                topic, UUID.randomUUID().toString(), json);

        try {
            RecordMetadata metadata = producer.send(producerRecord).get();
            System.out.println(String.format("================>send message result topic：%s", metadata.topic()));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
